package com.lkx.kafka.monitor.controller;

import com.lkx.kafka.monitor.dto.*;
import com.lkx.kafka.monitor.service.DataStorageService;
import com.lkx.kafka.monitor.service.KafkaMonitorService;
import lombok.RequiredArgsConstructor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;

import java.io.File;
import java.text.SimpleDateFormat;
import java.util.*;

@RestController
@RequestMapping("/kafka/metrics")
@RequiredArgsConstructor
@CrossOrigin(origins = "*") // 允许跨域访问，方便前端调试
public class KafkaMonitorController {

    @Autowired
    private KafkaMonitorService kafkaMonitorService;

    @Autowired
    private DataStorageService dataStorageService;

    /**
     * 获取集群整体指标
     */
    @GetMapping("/cluster")
    public ResponseEntity<ClusterMetrics> getClusterMetrics() {
        return ResponseEntity.ok(kafkaMonitorService.getClusterMetrics());
    }

    /**
     * 获取所有主题指标
     */
    @GetMapping("/topics")
    public ResponseEntity<List<TopicMetrics>> getAllTopicMetrics() {
        return ResponseEntity.ok(kafkaMonitorService.getAllTopicMetrics());
    }

    /**
     * 获取指定主题指标
     */
    @GetMapping("/topics/{topicName}")
    public ResponseEntity<TopicMetrics> getTopicMetrics(@PathVariable String topicName) {
        return ResponseEntity.ok(kafkaMonitorService.getTopicMetrics(topicName));
    }

    /**
     * 获取所有消费者组指标
     */
    @GetMapping("/consumer-groups")
    public ResponseEntity<List<ConsumerGroupMetrics>> getAllConsumerGroupMetrics() {
        return ResponseEntity.ok(kafkaMonitorService.getAllConsumerGroupMetrics());
    }

    /**
     * 获取指定消费者组指标
     */
    @GetMapping("/consumer-groups/{groupId}")
    public ResponseEntity<ConsumerGroupMetrics> getConsumerGroupMetrics(@PathVariable String groupId) {
        return ResponseEntity.ok(kafkaMonitorService.getConsumerGroupMetrics(groupId));
    }

    /**
     * 手动刷新指标数据
     */
    @PostMapping("/refresh")
    public ResponseEntity<Map<String, String>> refreshMetrics() {
        kafkaMonitorService.refreshMetrics();
        return ResponseEntity.ok(Map.of("status", "success", "message", "Metrics refreshed successfully"));
    }

    /**
     * 全局异常处理
     */
    @ExceptionHandler(IllegalArgumentException.class)
    public ResponseEntity<Map<String, String>> handleIllegalArgument(IllegalArgumentException e) {
        return ResponseEntity.status(HttpStatus.NOT_FOUND)
                .body(Map.of("status", "error", "message", e.getMessage()));
    }

    @ExceptionHandler(Exception.class)
    public ResponseEntity<Map<String, String>> handleException(Exception e) {
        return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR)
                .body(Map.of("status", "error", "message", "Failed to get metrics: " + e.getMessage()));
    }


    /**
     * 获取吞吐量指标
     * @param topicName 主题名称，可选
     * @param startTime 开始时间戳，可选
     * @param endTime 结束时间戳，可选
     * @param timeRange 时间范围(realtime, 1hour, 24hour, 7day)，可选
     * @return 吞吐量指标数据
     */
    @GetMapping("/throughput")
    public ResponseEntity<ThroughputMetricsDTO> getThroughputMetrics(
            @RequestParam(required = false) String topicName,
            @RequestParam(required = false) Long startTime,
            @RequestParam(required = false) Long endTime,
            @RequestParam(required = false, defaultValue = "24hour") String timeRange) {

        try {
            // 如果没有指定时间范围，则根据开始和结束时间计算
            if (startTime == null && endTime == null) {
                // 默认查询最近24小时数据
                endTime = System.currentTimeMillis();
                startTime = endTime - 24 * 60 * 60 * 1000L;

                // 根据timeRange调整时间范围
                switch (timeRange) {
                    case "realtime":
                        startTime = endTime - 5 * 60 * 1000L; // 最近5分钟
                        break;
                    case "1hour":
                        startTime = endTime - 60 * 60 * 1000L; // 最近1小时
                        break;
                    case "7day":
                        startTime = endTime - 7 * 24 * 60 * 60 * 1000L; // 最近7天
                        break;
                }
            }

            List<ThroughputMetrics> metricsList = dataStorageService.getData(topicName, startTime, endTime);

            ThroughputMetricsDTO result = convertToDTO(metricsList, topicName);
            return ResponseEntity.ok(result);
        } catch (Exception e) {
            return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
        }
    }

    private ThroughputMetricsDTO convertToDTO(List<ThroughputMetrics> metricsList, String topicName) {
        ThroughputMetricsDTO dto = new ThroughputMetricsDTO();
        dto.setTopicName(topicName);

        List<String> timestamps = new ArrayList<>();
        List<Long> produceRates = new ArrayList<>();
        List<Long> consumeRates = new ArrayList<>();

        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

        for (ThroughputMetrics metrics : metricsList) {
            timestamps.add(sdf.format(new Date(metrics.getTimestamp())));
            produceRates.add(metrics.getProduceRate() != null ? metrics.getProduceRate() : 0L);
            consumeRates.add(metrics.getConsumeRate() != null ? metrics.getConsumeRate() : 0L);
        }

        dto.setTimestamps(timestamps);
        dto.setProduceRates(produceRates);
        dto.setConsumeRates(consumeRates);

        return dto;
    }
}